package org.test.rocketmq.consumer01;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

/**
 * Push消费：并发消费
 */
public class PushConsumer {

    public static void main(String[] args) throws MQClientException {
        // 初始化consumer，并设置consumer group name
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("push_consumer_group");
        // 设置NameServer地址
        consumer.setNamesrvAddr("localhost:9876");
        //订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息
        consumer.subscribe("TopicTest", "*");
        // 设置集群模式或广播模式，默认为集群模式
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 设置并发度
        consumer.setConsumeThreadMin(10);
        // 设置最大并发度。这个值无效，因为使用的是无界队列
        consumer.setConsumeThreadMax(20);
        //注册回调接口来处理从Broker中收到的消息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
            for (MessageExt msg : msgs) {
                System.out.println("msgId:" + msg.getMsgId());
                System.out.println("body:" + new String(msg.getBody()));
            }
            // 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 启动Consumer
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
